use std::sync::mpsc;
use std::sync::mpsc::{Receiver,Sender};
use std::io;
use std::fs;
use std::io::BufRead;
use std::thread;
use regex::Regex;
use crate::common::Loc;

pub fn handle_each_line_async<F>(each: F,  loc: Loc)
where F: Fn(i32 , &str, Option<&Regex>, &Loc) + Send + Copy+ 'static
{
    
    let handler = io::stdin();
    // let mut stdlines:Vec<String>=  Vec::new();
    let mut lineno = 0;
    let  (tx, rx) = mpsc::channel::<(i32, String)>();
    let re = match Regex::new(&loc.re_string){
        Ok(r) =>{
            r
        },
        Err(e) => {
            println!("error compile {:?} || {}",loc.re_string, e.to_string());
            panic!(e)
        }
    };
    thread::spawn(move||{
        each_lines(rx, each, Some(&re), &loc)
    });

    // handle_thread.

    loop{
        let mut line = String::new(); 
        match handler.read_line(&mut line) { 
            Ok(n) => {
                if n == 0{
                    break;
                }
                tx.send((lineno, line.clone())).unwrap();
                lineno+=1;
            },
            Err(e) => {
                println!("{}",e);
                break;
            }
        }
    };
    tx.send((-1,"END".to_owned())).expect("error in end send");
    
}
pub struct OneWorker {
    tx :Sender<(i32,String)>,
    t: thread::JoinHandle<()>
}

impl OneWorker
{
    fn new<F>(each :F, regex_str:&str,loc: Loc) -> OneWorker
    where F: Fn(i32, &str, Option<&Regex>, &Loc) + Send + Copy+ 'static
    {
        let (tx, rx) = mpsc::channel::<(i32, String)>();
        let re = match Regex::new(regex_str){
            Ok(r) =>{
                r
            },
            Err(e) => {
                println!("error compile {:?} || {}",regex_str, e.to_string());
                panic!(e)
            }
        };
        
        OneWorker{
            tx:tx,
            t: thread::spawn(move||{
                each_lines(rx, each, Some(&re), &loc).expect("msg: &str");
            })
        }
    }
    fn join(self) -> io::Result<bool>{
        match self.t.join(){
            Ok(_) => {
                Ok(true)
            },
            Err(e) => {
                println!("{:#?}",e);
                Ok(false)
            }
        }
    }
    fn try_close(&self){
        self.tx.send((-1, "END".to_owned())).expect("msg: &str");
    }

    fn send_args(&self,i: i32, line: String){
        self.tx.send((i,line )).expect("send to thread error !");
    }
}

pub fn handle_each_line_file_async<F>(path :&str,each: F, loc :Loc, thread_num :i32) -> std::io::Result<()>
where F: Fn(i32, &str, Option<&Regex>, &Loc) + Send + Copy+ 'static
{
    let f =  fs::File::open(path)?;
    let buffer = io::BufReader::new(f);
    let mut channels = Vec::<OneWorker>::new();
    for _ in 0..thread_num{
        channels.push(OneWorker::new(each, &loc.re_string, loc.clone()))
    }
    let channels_len = channels.len();
    for (i,l) in buffer.lines().enumerate(){
        match l{
            Ok(line) => {
                channels[i % channels_len ].send_args(i as i32, line) ;
            },
            Err(_) => {
                break
            } 
            // None(_) => {}
        }
    }
    for work in channels.iter(){
        work.try_close();
    }
    for work in channels{
        work.join().unwrap();
    }

    Ok(())
}





pub fn each_lines<F>(r: Receiver<(i32,String)>, each :F, re :Option<&Regex>, loc:&Loc) -> io::Result<()>
where F: Fn(i32 , &str, Option<&Regex>, &Loc)
{
    loop{
        let (no,line) = r.recv().expect("recv error from channel");
        if no < 0 {
            break
        }
        each(no, &line, re, loc);
    }
    Ok(())
}